package com.winit.schedule.job.tracker.monitor;

import com.winit.schedule.job.tracker.repository.ScheduleDataManager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.winit.schedule.core.conf.Configuration;
import com.winit.schedule.core.utils.NetUtils;
import com.winit.schedule.core.zk.ZKService;
import com.winit.schedule.job.tracker.JobTrackerContextImpl;
import com.winit.schedule.job.tracker.conf.JobConfiguration;
import com.winit.schedule.job.tracker.event.ClusterEvent;
import com.winit.schedule.job.tracker.event.ClusterEventType;

import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

public class ZKServiceMonitor extends ZKService {

    private static transient final Logger logger  = LoggerFactory.getLogger(ZKServiceMonitor.class);

    private JobTrackerContextImpl         jobTrackerContext;

    private RegisterThread                registerThread;

    private BlackListThread               blackListThread;

    private ScheduleDataManager           scheduleDataManager;

    private String                        localIP = NetUtils.getLocalIP();

    public ZKServiceMonitor(JobTrackerContextImpl jobTrackerContext) throws Exception{
        super("ZKServiceMonitor");
        this.jobTrackerContext = jobTrackerContext;
        this.scheduleDataManager = jobTrackerContext.getScheduleDataManager();
    }

    @SuppressWarnings("unchecked")
    @Override
    protected void serviceInit(Configuration conf) throws Exception {
        super.serviceInit(conf);
        this.addListener(new ClusterListener() {

            @Override
            public void masterChanage(boolean isMaster, String masterAddress) {
                if (isMaster) {
                    logger.info("------------JobTracker is master!!!");
                } else {
                    logger.info("------------JobTracker is slave!!!");
                }
                logger.info("------------master is : " + masterAddress);

                getJobTrackerContext().getDispatcher()
                    .getEventHandler()
                    .handle(new ClusterEvent(ClusterEventType.MASTER_CHANAE, isMaster));
            }

            @Override
            public void disconnected() {
                getJobTrackerContext().getDispatcher()
                    .getEventHandler()
                    .handle(new ClusterEvent(ClusterEventType.MASTER_CHANAE, false));
            }

            @Override
            public void connected() {
                // 线程定时检查并注册节点信息（zookeeper会话异常中断补偿手段）
                if (null != registerThread) {
                    registerThread.cancel();
                }

                registerThread = new RegisterThread();
                registerThread.start();

                // 线程定时拉取黑名单信息（zookeeper会话异常中断补偿手段）
                if (null != blackListThread) {
                    blackListThread.cancel();
                }

                blackListThread = new BlackListThread();
                blackListThread.start();
            }
        });

        // 添加白名单监听
        this.addBlackListListener(new BlackListListener() {

            @Override public void chanaged(String blacklist) {
                scheduleDataManager.resetBlackListCache(blacklist);
            }
        });
    }

    @Override
    protected void serviceStart() throws Exception {
        super.serviceStart();

        long whiteListCacheRefreshIntervalS = jobTrackerContext.getConf()
                .getLong(JobConfiguration.WHITELIST_CACHE_REFRESH_INVERVAL,
                        JobConfiguration.DEFAULT_WHITELIST_CACHE_REFRESH_INVERVAL);
    }

    @Override
    protected void serviceStop() throws Exception {
        super.serviceStop();
    }

    @Override
    protected void reRegisterJobTracker() {

        try {
            List<String> childrens = this.getZooKeeper().getChildren(this.getPathServer(), false);
            if (null == childrens || childrens.size() == 0) {
                boolean develoyMode = getConfig().getBoolean(JobConfiguration.DEVELOP_MODE,
                        JobConfiguration.DEFAULT_DEVELOP_MODE);
                if (!develoyMode) {
                    registerJobTracker(localIP,
                            getConfig().getInt(JobConfiguration.LISTEN_PORT, JobConfiguration.DEFAULT_LISTEN_PORT));
                } else {
                    registerJobTracker_dev(localIP,
                            getConfig().getInt(JobConfiguration.LISTEN_PORT, JobConfiguration.DEFAULT_LISTEN_PORT));
                }
            }


        } catch (InterruptedException e) {
            logger.warn("-------- Jobtracker failed to register node with zookeeper", e);
        } catch (Exception e) {
            logger.warn("-------- Jobtracker failed to register node with zookeeper", e);
        }
    }

    public JobTrackerContextImpl getJobTrackerContext() {
        return jobTrackerContext;
    }

    class RegisterThread extends Thread {

        private boolean needRegister = true;

        @SuppressWarnings("static-access")
        @Override
        public void run() {
            while (needRegister) {
                try {
                    Thread.currentThread().sleep(3000);

                    if (needRegister) {
                        boolean develoyMode = getConfig().getBoolean(JobConfiguration.DEVELOP_MODE,
                                JobConfiguration.DEFAULT_DEVELOP_MODE);
                        if (!develoyMode) {
                            registerJobTracker(localIP,
                                    getConfig().getInt(JobConfiguration.LISTEN_PORT, JobConfiguration.DEFAULT_LISTEN_PORT));
                        } else {
                            registerJobTracker_dev(localIP,
                                    getConfig().getInt(JobConfiguration.LISTEN_PORT, JobConfiguration.DEFAULT_LISTEN_PORT));
                        }

                        needRegister = false;
                        logger.info("-------- Jobtracker success to register node with zookeeper");
                    }

                } catch (InterruptedException e) {
                    logger.warn("-------- Jobtracker failed to register node with zookeeper", e);
                } catch (Exception e) {
                    logger.warn("-------- Jobtracker failed to register node with zookeeper", e);
                }
            }
        }

        public boolean isRunning() {
            return needRegister;
        }

        public void cancel() {
            needRegister = false;
            Thread.currentThread().interrupt();
        }

    }

    class BlackListThread extends Thread {

        private boolean needRegister = true;

        private String currentBlackList = "";

        @Override public void run() {
            try {
                while (needRegister) {
                    Thread.currentThread().sleep(60000);

                    String blackList = getBlackList();

                    scheduleDataManager.resetBlackListCache(blackList);
                    logger.info("-------- Time refresh blackList");
                    if (!currentBlackList.equals(blackList)) {
                        currentBlackList = blackList;
                        logger.info("-------- BlackList change : " + blackList);
                    }
                }
            } catch (InterruptedException e) {
                logger.warn("-------- Time refresh blacklist", e);
            } catch (Exception e) {
                logger.warn("-------- Time refresh blacklist", e);
            }
        }

        public boolean isRunning() {
            return needRegister;
        }

        public void cancel() {
            needRegister = false;
            Thread.currentThread().interrupt();
        }
    }
}
